package com.nl.orderpay;

import com.nl.bean.input.OrderEvent;
import com.nl.bean.input.ReceiptEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author shihb
 * @date 2019/12/27 10:17
 * 对账类，联合多条流 一条订单支付，一条支付回执
 */
public class TransactionReconciliation {

  /**
   * 定义侧输出流tag
   */
  static OutputTag<OrderEvent> missReceipt = new OutputTag<OrderEvent>("missReceipt"){};
  static OutputTag<ReceiptEvent> missPay = new OutputTag<ReceiptEvent>("missPay"){};

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    //1.订单事件流
    DataStreamSource<String> paySource = env.readTextFile(TransactionReconciliation.class.getClassLoader().getResource("OrderLog.csv").getPath());
    //2.交易回执流
    DataStreamSource<String> receiptSource = env.readTextFile(TransactionReconciliation.class.getClassLoader().getResource("ReceiptLog.csv").getPath());

    KeyedStream<OrderEvent, String> payStream = paySource
        .map(s -> {
          String[] arr = s.split(",");
          return new OrderEvent(Long.parseLong(arr[0].trim()), arr[1].trim(), arr[2].trim(),
              Long.parseLong(arr[3].trim()));
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<OrderEvent>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(OrderEvent element) {
                return element.getEventTime() * 1000;
              }
            }
        )
        .filter(orderEvent -> !"".equals(orderEvent.getTransactionId()))
        .keyBy(orderEvent -> orderEvent.getTransactionId());

    KeyedStream<ReceiptEvent, String> receiptStream = receiptSource
        .map(s->{
          String[] arr = s.split(",");
          return new ReceiptEvent(arr[0].trim(), arr[1].trim(), Long.parseLong(arr[2].trim()));
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<ReceiptEvent>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(ReceiptEvent element) {
                return element.getReceiptTime() * 1000;
              }
            }
        ).keyBy(ReceiptEvent->ReceiptEvent.getTransactionId());

    // 将两条流连接起来,共同处理
    SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> trStream = payStream
        .connect(receiptStream).process(new TrFunction());

    trStream.print("success");
    trStream.getSideOutput(missPay).print("miss pay");
    trStream.getSideOutput(missReceipt).print("miss receipt");
    env.execute("tr job");

  }

  /**
   * 自定义CoProcessFunction
   * IN1.connect(IN2)
   * IN1:输入流1的数据类型
   * IN2:输入流2的数据类型
   * OUT:输出流的数据类型
   */
  private static class TrFunction extends CoProcessFunction<OrderEvent,ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>>{

    private ValueState<OrderEvent> payState;
    private ValueState<ReceiptEvent> receiptState;

    @Override
    public void open(Configuration parameters) throws Exception {
      // 定义状态保存到来的事件
      payState=getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("payState",OrderEvent.class));
      receiptState=getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receiptState",ReceiptEvent.class));
    }

    /**
     * 第1条流的处理，这里是订单支付流
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement1(OrderEvent value, Context ctx,
        Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
      ReceiptEvent curReceipt = receiptState.value();
      if(curReceipt!=null){
        //到账的已经来了，完成对账，发送清空状态
        out.collect(new Tuple2<OrderEvent,ReceiptEvent>(value,curReceipt));
        this.receiptState.clear();
      }else{
        //没到账信息就更新状态,注册定时器等一下
        this.payState.update(value);
        //定时器的时间要根据两条流分析确定
        ctx.timerService().registerEventTimeTimer(value.getEventTime()*1000L+5000);
      }
    }

    /**
     * 第2条流的处理，当前方法是支付回执流
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement2(ReceiptEvent value, Context ctx,
        Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
      OrderEvent curPay = this.payState.value();
      if(curPay!=null){
        //支付信息已经来了，完成对账
        out.collect(new Tuple2<OrderEvent,ReceiptEvent>(curPay,value));
        this.payState.clear();
      }else{
        //没到支付信息就更新状态,注册定时器等一下
        this.receiptState.update(value);
        //定时器的时间要根据两条流分析确定
        ctx.timerService().registerEventTimeTimer(value.getReceiptTime()*1000L+5000);
      }
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
        Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
      OrderEvent curPay = this.payState.value();
      ReceiptEvent curReceipt = receiptState.value();
      if(curPay!=null && curReceipt!=null){
        //对账成功
        out.collect(new Tuple2<OrderEvent,ReceiptEvent>(curPay,curReceipt));
      }else if(curPay==null && curReceipt!=null){
        //有对账信息但没pay
        ctx.output(missPay,curReceipt);
      }else if(curPay!=null && curReceipt==null){
        //有pay，没对账信息
        ctx.output(missReceipt,curPay);
      }
      this.payState.clear();
      this.receiptState.clear();
    }
  }
}
